package report

import java.util.Properties

import Configer.Config
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

//用sparkSQL实现地区统计
object AreaKpiSQL {
  def main(args: Array[String]): Unit = {
    //sparkContext
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getName}")
    conf.setMaster("local[*]")
    conf.set("spark.serializer",Config.serializer)
    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)
    //读取数据
    val dataFrame = sQLContext.read.parquet(Config.parquetPath)

    sQLContext.udf.register("xiaoniu",(a:Boolean)=>if (a) 1 else 0)
    //分析数据
    dataFrame.registerTempTable("area")
    val result = sQLContext.sql(
      """
        |select provincename,cityname,
        |sum(xiaoniu(requestmode=1 and	processnode>=1)) as adoldreq,
        |sum(case when requestmode=1 and	processnode>=2 then 1 else 0 end) as adeffreq,
        |sum(if(requestmode=1 and	processnode>=3,1,0)) as adreq,
        |sum(if(iseffective=1 and	isbilling=1 and isbid=1 and	adorderid !=0,1,0)) as adrtbreq,
        |sum(if(iseffective=1 and	isbilling=1 and iswin=1,1,0)) as adsucreq,
        |sum(if(requestmode=2 and	iseffective=1,1,0)) as adshow,
        |sum(if(requestmode=3 and	iseffective=1,1,0)) as adclick,
        |sum(if(iseffective=1 and	isbilling=1 and iswin=1,adpayment/1000,0)) as adadpreq,
        |sum(if(iseffective=1 and	isbilling=1 and iswin=1,winprice/1000,0)) as adcusreq
        |from area group by provincename,cityname
      """.stripMargin)

    //存储数据
    val props = new Properties()
    props.setProperty("driver",Config.driver)
    props.setProperty("user",Config.user)
    props.setProperty("password",Config.password)
    result.write.jdbc(Config.url,Config.table,props)

    //释放资源
    sc.stop()
  }
}
